/*
* Copyright(C) 2011-2012 Alibaba Group Holding Limited
* 
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation.
*/


#include <easy_string.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <pthread.h>
#include "easy_io.h"
#include "easy_connection.h"
#include "easy_message.h"
#include "easy_request.h"
#include "easy_file.h"
#include "easy_client.h"
#include "easy_socket.h"
#include "easy_log.h"

static easy_connection_t *easy_connection_new();
static void easy_connection_on_accept(struct ev_loop *loop, ev_io *w, int revents);
static void easy_connection_on_readable(struct ev_loop *loop, ev_io *w, int revents);
static void easy_connection_on_writable(struct ev_loop *loop, ev_io *w, int revents);
static void easy_connection_on_timeout_mesg(struct ev_loop *loop, ev_timer *w, int revents);
static void easy_connection_on_timeout_conn(struct ev_loop *loop, ev_timer *w, int revents);
static void easy_connection_redispatch_thread(easy_connection_t *c);
static void easy_connection_evio_start(easy_connection_t *c);

static uint64_t easy_connection_get_packet_id(easy_connection_t *c, void *packet, int flag);
static int easy_connection_do_request(easy_message_t *m);
static int easy_connection_do_response(easy_message_t *m);
static int easy_connection_send_response(easy_list_t *request_list);
static void easy_connection_add_list(easy_connection_list_t *wlist, easy_connection_t *c);

static int easy_connection_write_again(easy_connection_t *c);
static easy_message_t *easy_connection_recycle_message(easy_message_t *m);
static easy_connection_t *easy_connection_do_connect(easy_client_t *client);
static easy_connection_t *easy_connection_do_client(easy_session_t *s);
static void easy_connection_autoconn(easy_connection_t *c);

/**
 * 增加监听端口, 要在easy_io_start开始调用
 *
 * @param host  机器名或IP, 或NULL
 * @param port  端口号
 *
 * @return      如果成功返回easy_connection_t对象, 否则返回NULL
 */
easy_listen_t *easy_connection_add_listen(easy_io_t *eio,
        const char *host, int port, easy_io_handler_pt *handler)
{
    easy_addr_t address;

    if ((address = easy_inet_str_to_addr(host, port)).addr == 0) {
        easy_trace_log("error addr: host=%s, port=%d.\n", host, port);
        return NULL;
    }

    return easy_connection_listen_addr(eio, address, handler);
}
// 通过easy_addr_t
easy_listen_t *easy_connection_listen_addr(easy_io_t *eio, easy_addr_t addr, easy_io_handler_pt *handler)
{
    int                     i, size, cnt;
    easy_listen_t           *l;

    // 已经开始,不能再增加监听
    if (eio->started || eio->pool == NULL) {
        easy_error_log("easy_connection_add_listen failure: eio->started=%d, eio->pool=%p\n",
                       eio->started, eio->pool);
        return NULL;
    }

    // alloc memory
    cnt = (eio->listen_all || eio->io_thread_count == 1) ? eio->io_thread_count : 2;
    size = cnt * sizeof(ev_io);
    size += sizeof(easy_listen_t);

    if ((l = (easy_listen_t *) easy_pool_calloc(eio->pool, size)) == NULL) {
        easy_error_log("easy_pool_calloc failure: eio->pool=%p, size=%d\n",
                       eio->pool, size);
        return NULL;
    }

    // 打开监听
    l->addr = addr;

    if ((l->fd = easy_socket_listen(&l->addr)) < 0) {
        char buffer[32];
        easy_error_log("easy_socket_listen failure: host=%s\n",
                       easy_inet_addr_to_str(&l->addr, buffer, 32));
        return NULL;
    }

    // 初始化
    l->handler = handler;

    for(i = 0; i < cnt; i++) {
        ev_io_init(&l->read_watcher[i], easy_connection_on_accept, l->fd, EV_READ | EV_CLEANUP);
        ev_set_priority (&l->read_watcher[i], EV_MAXPRI);
        l->read_watcher[i].data = l;
    }

    l->next = eio->listen;
    eio->listen = l;

    return l;
}

/**
 * session_timeout
 */
void easy_connection_wakeup_session(easy_connection_t *c)
{
    easy_session_t  *s, *sn;

    if (c->send_queue) {
        easy_list_for_each_entry_safe(s, sn, &(c->send_queue->list), send_queue_list) {
            easy_hash_del_node(&s->send_queue_hash);
            easy_session_process(s, 1);
        }
        easy_list_init(&c->send_queue->list);
    }
}

/**
 * destroy掉easy_connection_t对象
 *
 * @param c - easy_connection_t对象
 */
void easy_connection_destroy(easy_connection_t *c)
{
    easy_message_t  *m, *m2;
    easy_io_t       *eio;

    // release session
    easy_connection_wakeup_session(c);

    // disconnect
    eio = c->ioth->eio;

    if (c->status != EASY_CONN_CLOSE && c->handler->on_disconnect) {
        (c->handler->on_disconnect)(c);
    }

    // refcount
    if (c->status != EASY_CONN_CLOSE && c->pool->ref > 0 && eio->stoped == 0) {
        ev_io_stop(c->loop, &c->read_watcher);
        ev_io_stop(c->loop, &c->write_watcher);
        c->status = EASY_CONN_CLOSE;

        // wakeup other thread
        if (c->type == EASY_TYPE_SERVER) {
            easy_thread_pool_t *tp = eio->thread_pool;

            while (tp) {
                easy_baseth_pool_on_wakeup(tp);
                tp = tp->next;
            }
        }

        if (c->pool->ref > 0) {
            ev_timer_set(&c->timeout_watcher, 0.0, 0.5);
            ev_timer_again(c->loop, &c->timeout_watcher);
        }
    }

    if (c->type == EASY_TYPE_SERVER) {
           easy_request_t *r, *n;
           easy_list_for_each_entry_safe(r, n, &c->session_list, request_list_node) {
               easy_request_server_done(r);
           }
       }


    if (c->pool->ref > 0 && eio->stoped == 0) return;

    // release request
    if (c->type == EASY_TYPE_SERVER) {
        easy_request_t *r, *n;
        easy_list_for_each_entry_safe(r, n, &c->session_list, request_list_node) {
            easy_list_del(&r->request_list_node);
        }
    }

    // release message
    easy_list_for_each_entry_safe(m, m2, &c->message_list, message_list_node) {
        easy_message_destroy(m, 1);
    }
    easy_buf_chain_clear(&c->output);
    ev_io_stop(c->loop, &c->read_watcher);
    ev_io_stop(c->loop, &c->write_watcher);
    ev_timer_stop(c->loop, &c->timeout_watcher);

    // close
    if (c->fd >= 0) {
        easy_debug_log("[%p] connection close: %d\n", c, c->fd);

        if (!c->read_eof) {
            char buf[EASY_POOL_PAGE_SIZE];

            while (read(c->fd, buf, EASY_POOL_PAGE_SIZE) > 0);
        }

        close(c->fd);
        c->fd = -1;
    }

    // autoreconn
    if (c->auto_reconn && eio->stoped == 0) {
        c->status = EASY_CONN_AUTO_CONN;
        double t = c->reconn_time / 1000.0 * (1 << c->reconn_fail);

        if (t > 15) t = 15;

        if (c->reconn_fail < 16) c->reconn_fail ++;

        easy_warn_log("[%p] connection reconn_time: %f, reconn_fail: %d\n", c, t, c->reconn_fail);
        ev_timer_set(&c->timeout_watcher, 0.0, t);
        ev_timer_again(c->loop, &c->timeout_watcher);
        return;
    }

    easy_list_del(&c->conn_list_node);

    if (c->client) c->client->c = NULL;

    if (eio->stoped) c->pool->ref = 0;

    easy_pool_destroy(c->pool);
}

int easy_connection_connect(easy_io_t *eio, easy_addr_t addr,
                            easy_io_handler_pt *handler, int conn_timeout, void *args, int autoconn)
{
    int         ret;

    if (addr.addr == 0)
        return EASY_ERROR;

    easy_session_t *s = easy_session_create(0);
    s->status = EASY_CONNECT_ADDR;
    s->process = (easy_io_process_pt *)handler;
    s->timeout = conn_timeout;
    s->r.args = args;
    s->packet_id = autoconn;

    if ((ret = easy_client_dispatch(eio, addr, s)) != EASY_OK)
        easy_session_destroy(s);

    return ret;
}

easy_connection_t *easy_connection_connect_thread(easy_io_t *eio, easy_addr_t addr,
        easy_io_handler_pt *handler, int conn_timeout, void *args, int autoconn)
{
    if (addr.addr == 0)
        return NULL;

    easy_session_t s;
    memset(&s, 0, sizeof(easy_session_t));
    s.status = EASY_CONNECT_ADDR;
    s.addr = addr;
    s.process = (easy_io_process_pt *)handler;
    s.timeout = conn_timeout;
    s.r.args = args;
    s.packet_id = autoconn;
    return easy_connection_do_client(&s);
}

int easy_connection_disconnect(easy_io_t *eio, easy_addr_t addr)
{
    int             ret;

    if (addr.addr == 0)
        return EASY_ERROR;

    easy_session_t *s = easy_session_create(0);
    s->status = EASY_DISCONNECT_ADDR;

    if ((ret = easy_client_dispatch(eio, addr, s)) != EASY_OK) {
        easy_session_destroy(s);
    }

    return ret;
}

int easy_connection_disconnect_thread(easy_io_t *eio, easy_addr_t addr)
{
    if (addr.addr == 0)
        return EASY_ERROR;

    easy_session_t s;
    memset(&s, 0, sizeof(easy_session_t));
    s.status = EASY_DISCONNECT_ADDR;
    s.addr = addr;
    easy_connection_do_client(&s);
    return EASY_OK;
}

int easy_connection_session_build(easy_session_t *s)
{
    double                  t;
    easy_connection_t       *c = s->c;

    if (c->type != EASY_TYPE_CLIENT)
        return EASY_ERROR;

    // 得到packet_id
    s->packet_id = easy_connection_get_packet_id(c, s->r.opacket, 0);

    // encode
    (c->handler->encode)(&s->r, s->r.opacket);

    s->timeout_watcher.data = s;
    easy_hash_dlist_add(c->send_queue, s->packet_id, &s->send_queue_hash, &s->send_queue_list);
    c->pool->ref ++;
    c->doing_request_count ++;

    // timeout设置
    t = (s->timeout ? s->timeout : EASY_CLIENT_DEFAULT_TIMEOUT) / 1000.0;

    if (c->status == EASY_CONN_OK) t += ev_now(c->loop);

    ev_timer_init(&s->timeout_watcher, easy_connection_on_timeout_mesg, t, 0.0);

    // 加入c->session_list
    if (c->status != EASY_CONN_OK) {
        ev_timer_start(c->loop, &s->timeout_watcher);
    } else {
        easy_list_add_tail(&s->session_list_node, &c->session_list);
        ev_timer_set(&c->timeout_watcher, 0.0, EASY_MIN_INTERVAL);
        ev_timer_again(c->loop, &c->timeout_watcher);
    }

    return EASY_OK;
}

/**
 * 发送到c上, 只允许本io线程调用
 */
int easy_connection_send_session(easy_connection_t *c, easy_session_t *s)
{
    easy_io_thread_t        *ioth = EASY_IOTH_SELF;

    if (ioth == NULL || ioth->iot == 0 || ioth->eio->stoped) {
        return EASY_ERROR;
    }

    if (s->status == 0 && ioth->doing_request_count >= EASY_IOTH_DOING_REQ_CNT) {
        easy_error_log("ioth->doing_request_count: %d, EASY_IOTH_DOING_REQ_CNT: %d\n",
                       ioth->doing_request_count, EASY_IOTH_DOING_REQ_CNT);
        return EASY_ERROR;
    }

    s->c = c;

    if (s->process == NULL) s->process = c->handler->process;

    easy_atomic32_inc(&ioth->doing_request_count);

    if (easy_connection_session_build(s) != EASY_OK)
        return EASY_ERROR;

    // 等下次写出去
    if (c->status != EASY_CONN_OK || ev_is_active(&c->write_watcher))
        return EASY_OK;

    // 写出到socket
    if (easy_connection_write_socket(c) == EASY_ABORT) {
        easy_connection_destroy(c);
        return EASY_ABORT;
    }

    return EASY_OK;
}
///////////////////////////////////////////////////////////////////////////////////////////////////
/**
 * accept 事件处理
 */
static void easy_connection_on_accept(struct ev_loop *loop, ev_io *w, int revents)
{
    int                     fd;
    easy_listen_t           *listen;
    struct sockaddr_in      addr;
    socklen_t               addr_len;
    easy_connection_t       *c;
    easy_io_thread_t        *ioth;
    char                    buffer[32];

    listen = (easy_listen_t *) w->data;
    addr_len = sizeof(struct sockaddr_in);
    assert(w->fd == listen->fd);

    // accept
    if ((fd = accept(w->fd, (struct sockaddr *)&addr, &addr_len)) < 0) return;

    // 为新连接创建一个easy_connection_t对象
    if ((c = easy_connection_new()) == NULL) {
        easy_error_log("easy_connection_new\n");
        close(fd);
        return;
    }

    // 设为非阻塞
    easy_socket_non_blocking(fd);

    // 初始化
    c->fd = fd;
    c->type = EASY_TYPE_SERVER;
    c->handler = listen->handler;
    c->addr = *((easy_addr_t *)&addr);

    // 事件初始化
    ev_io_init(&c->read_watcher, easy_connection_on_readable, fd, EV_READ);
    ev_io_init(&c->write_watcher, easy_connection_on_writable, fd, EV_WRITE);
    ev_init(&c->timeout_watcher, easy_connection_on_timeout_conn);
    c->read_watcher.data = c;
    c->write_watcher.data = c;
    c->timeout_watcher.data = c;
    c->ioth = ioth = (easy_io_thread_t *)easy_baseth_self;
    c->loop = loop;
    c->start_time = ev_now(ioth->loop);

    easy_debug_log("accpet from '%s', connection=%p, fd=%d\n",
                   easy_inet_addr_to_str(&c->addr, buffer, 32), c, c->fd);

    // on connect
    if (c->handler->on_connect)
        (c->handler->on_connect)(c);

    // start idle
    if (c->handler->on_idle) {
        double t = easy_max(1.0, c->idle_time / 2000.0);
        ev_timer_set(&c->timeout_watcher, 0.0, t);
        ev_timer_again(c->loop, &c->timeout_watcher);
    }


    easy_atomic_add(&ioth->eio->connect_num, 1);
    int th_idx = (ioth->eio->connect_num) % (ioth->eio->io_thread_count);
    th_idx = th_idx<0 ? (0-th_idx) : th_idx;
    // 让出来给其他的线程
    if (ioth->eio->listen_all == 0 && listen->old_ioth == NULL
            && listen->curr_ioth == ioth) {
        easy_io_thread_t *nextth;
        listen->old = listen->cur;
        listen->curr_ioth = NULL;
        //listen->old_ioth = ioth;
        //ioth->listen_watcher.repeat = 0.5;
        ev_io_stop(ioth->loop, &listen->read_watcher[listen->old]);
        listen->old_ioth = NULL;
        nextth = easy_thread_pool_index(ioth->eio->io_thread_pool,th_idx);
        easy_unlock(&listen->listen_lock);

        ev_async_send(nextth->loop, &nextth->listen_watcher);

        //ev_timer_again (ioth->loop, &ioth->listen_watcher);

    }

    // start read
    easy_list_add_tail(&c->conn_list_node, &c->ioth->connected_list);
    c->event_status = EASY_EVENT_READ;
    easy_connection_on_readable(loop, &c->read_watcher, 0);

    return;
}

static void easy_connection_evio_start(easy_connection_t *c)
{
    if ((c->event_status & EASY_EVENT_READ))
        ev_io_start(c->loop, &c->read_watcher);

    if ((c->event_status & EASY_EVENT_WRITE))
        ev_io_start(c->loop, &c->write_watcher);

    if ((c->event_status & EASY_EVENT_TIMEOUT))
        ev_timer_start(c->loop, &c->timeout_watcher);

    c->event_status = 0;
}

/**
 * 为了均衡，切换到其他线程
 */
static void easy_connection_redispatch_thread(easy_connection_t *c)
{
    easy_io_thread_t    *ioth;

    // 处理了８次以下, 或者有读写没完, 不能切换
    if (!c->need_redispatch || easy_list_empty(&c->message_list) == 0 ||
            easy_list_empty(&c->output) == 0 || EASY_IOTH_SELF->eio->no_redispatch)
        return;

    // 选择一新的ioth
    ioth = (easy_io_thread_t *)easy_thread_pool_rr(EASY_IOTH_SELF->eio->io_thread_pool, 0);

    if (ioth == c->ioth)
        return;

    easy_list_del(&c->conn_list_node);
    ev_io_stop(c->loop, &c->read_watcher);
    ev_io_stop(c->loop, &c->write_watcher);
    ev_timer_stop(c->loop, &c->timeout_watcher);

    easy_info_log("connetion=%p redispatch %p to %p\n", c, c->ioth, ioth);

    // 加入到新的队列中
    c->event_status = EASY_EVENT_READ;
    c->ioth = ioth;
    c->loop = ioth->loop;
    c->need_redispatch = 0;

    easy_spin_lock(&ioth->thread_lock);
    easy_list_add_tail(&c->conn_list_node, &ioth->conn_list);
    ev_async_send(ioth->loop, &ioth->thread_watcher);
    easy_spin_unlock(&ioth->thread_lock);
}

/**
 * 切换listen
 */
void easy_connection_on_listen(struct ev_loop *loop, ev_async *w, int revents)
{
    easy_listen_t               *l;
    easy_io_thread_t            *ioth;
    ioth = (easy_io_thread_t *) w->data;

    // 对每一个listen
    for (l = ioth->eio->listen; l; l = l->next) {
        // trylock一下
        if (easy_trylock(&l->listen_lock)) {
            // 是自己
            if (l->old_ioth == ioth) {
                l->old_ioth = NULL;
                l->curr_ioth = ioth;
            } else {
                l->cur = ((l->cur + 1) & 1);
                ev_io_start(ioth->loop, &l->read_watcher[l->cur]);
                l->curr_ioth = ioth;
//                ioth->listen_watcher.repeat = 60.;
//                ev_timer_again (ioth->loop, &ioth->listen_watcher);
            }
        }
    }
}

/**
 * conn事件处理
 */
void easy_connection_on_wakeup(struct ev_loop *loop, ev_async *w, int revents)
{
    easy_connection_t   *c;
    easy_io_thread_t    *ioth;
    easy_list_t         conn_list;
    easy_list_t         session_list;
    easy_list_t         request_list;

    ioth = (easy_io_thread_t *) w->data;

    // 取回list
    easy_spin_lock(&ioth->thread_lock);
    easy_list_movelist(&ioth->conn_list, &conn_list);
    easy_list_movelist(&ioth->session_list, &session_list);
    easy_list_movelist(&ioth->request_list, &request_list);
    easy_spin_unlock(&ioth->thread_lock);

    // foreach
    easy_list_for_each_entry(c, &conn_list, conn_list_node) {
        c->loop = loop;
        c->start_time = ev_now(ioth->loop);
        easy_connection_evio_start(c);
    }
    easy_list_join(&conn_list, &ioth->connected_list);

    easy_connection_send_session_list(&session_list);
    easy_connection_send_response(&request_list);
}

/**
 * read事件处理
 */
static void easy_connection_on_readable(struct ev_loop *loop, ev_io *w, int revents)
{
    easy_connection_t   *c;
    easy_message_t      *m;
    int                 n;

    c = (easy_connection_t *)w->data;
    assert(c->fd == w->fd);

    // 防止请求过多
    if (c->type == EASY_TYPE_SERVER && (c->doing_request_count > EASY_CONN_DOING_REQ_CNT ||
                                        c->ioth->doing_request_count > EASY_IOTH_DOING_REQ_CNT)) {
        easy_warn_log("c->doing_request_count: %d, c->ioth->doing_request_count: %d\n",
                      c->doing_request_count, c->ioth->doing_request_count);
        goto error_exit;
    }

    // 最后的请求, 如果数据没完, 需要继续读
    m = easy_list_get_last(&c->message_list, easy_message_t, message_list_node);

    // 第一次读或者上次读完整了, 重新建一个easy_message_t
    if (m == NULL || m->status != EASY_MESG_READ_AGAIN) {
        if ((m = easy_message_create(c)) == NULL) {
            easy_error_log("easy_message_create failure, c=%p\n", c);
            goto error_exit;
        }
    }

    // 检查buffer大小
    if (easy_buf_check_read_space(m->pool, m->input, m->next_read_len) != EASY_OK) {
        easy_error_log("easy_buf_check_read_space failure, m=%p, len=%d\n", m, m->next_read_len);
        goto error_exit;
    }

    // 从conn里读入数据
    if ((n = read(c->fd, m->input->last, m->next_read_len)) <= 0) {
        easy_debug_log("n: %d, errno: %s(%d)\n", n, strerror(errno), errno);
        goto error_exit;
    }

    c->read_eof = (n < m->next_read_len);
    c->last_time = ev_now(loop);
    c->reconn_fail = 0;

    easy_debug_log("read: %d, fd: %d\n", n, c->fd);

    if (easy_log_level >= EASY_LOG_TRACE) {
        char btmp[128];
        easy_trace_log("read: %d => %s", n, easy_string_tohex(m->input->last, n, btmp, 128));
    }

    easy_atomic_add(&EASY_IOTH_SELF->eio->recv_byte, n);
    m->input->last += n;

    if (c->default_message_len < EASY_IO_BUFFER_SIZE && m->next_read_len == n)
        c->default_message_len = EASY_IO_BUFFER_SIZE;

    // client
    if (EASY_ERROR == ((c->type == EASY_TYPE_CLIENT) ?
                       easy_connection_do_response(m) : easy_connection_do_request(m))) {
        easy_debug_log("c->type=%d, fd=%d\n", c->type, c->fd);
        goto error_exit;
    }

    return;
error_exit:
    easy_connection_destroy(c);
}

/**
 * 处理响应
 */
static int easy_connection_do_response(easy_message_t *m)
{
    easy_connection_t       *c;
    easy_session_t          *s;
    uint64_t                packet_id;
    int                     i, cnt, left;
    void                    *packet;
    easy_list_t             list;

    c = m->c;

    // 处理buf
    cnt = 0;
    left = 0;
    easy_list_init(&list);

    while (m->input->pos < m->input->last) {
        if ((packet = (c->handler->decode)(m)) == NULL) {
            if (m->status != EASY_ERROR)
                break;

            easy_warn_log("decode error, fd=%d\n", c->fd);
            return EASY_ERROR;
        }

        cnt ++;
        packet_id = easy_connection_get_packet_id(c, packet, 1);
        s = (easy_session_t *) easy_hash_dlist_del(c->send_queue, packet_id);

        if (s == NULL) {
            // 需要cleanup
            if (c->handler->cleanup)
                (c->handler->cleanup)(NULL, packet);

            easy_warn_log("not found session, c=%p, packet_id=%ld\n", c, packet_id);
            continue;
        }

        // process
        EASY_IOTH_SELF->done_request_count ++;
        s->r.ipacket = packet;              // in

        if (s->async) {                     // message延后释放
            m->async = s->async;
            easy_atomic_inc(&m->pool->ref);
            s->r.request_list_node.next = (easy_list_t *) m;
        }

        // stop timer
        ev_timer_stop(c->loop, &s->timeout_watcher);
        easy_list_del(&s->session_list_node);
        easy_request_client_done(&s->r);
        c->pool->ref --;

        if (c->handler->batch_process) {
            easy_list_add_tail(&s->session_list_node, &list);

            if (++ left >= 32) {
                (c->handler->batch_process)((easy_message_t *)&list);
                left = 0;
            }
        } else if (easy_session_process(s, 0) == EASY_ERROR) {
            easy_warn_log("easy_session_process error, fd=%d, s=%p\n", c->fd, s);
            return EASY_ERROR;
        }
    }

    // batch process
    if (cnt) m->recycle_cnt ++;

    if (left > 0) {
        (c->handler->batch_process)((easy_message_t *)&list);
    }

    // close
    if (c->wait_close && c->pool->ref == 0) {
        return EASY_ERROR;
    }

    // send new packet
    if (c->handler->new_packet) {
        left = (EASY_CONN_DOING_REQ_CNT >> 1) - c->doing_request_count;

        if (c->ioth->doing_request_count > 0) {
            left = easy_min(left, (EASY_IOTH_DOING_REQ_CNT >> 1) - c->ioth->doing_request_count);
        }

        if (left > 0) {
            ev_io_start(c->loop, &c->write_watcher);
            left = easy_min(cnt, left);

            for(i = 0; i < left; i++) {
                if ((c->handler->new_packet)(c) == EASY_ERROR)
                    return EASY_ERROR;
            }
        }
    }

    if ((m = easy_connection_recycle_message(m)) == NULL) {
        easy_warn_log("easy_connection_recycle_message error, fd=%d, m=%p\n", c->fd, m);
        return EASY_ERROR;
    }

    // status, message 没读完
    if (m->input->pos < m->input->last) {
        m->status = EASY_MESG_READ_AGAIN;
    } else {
        easy_message_destroy(m, 1);
    }

    return EASY_OK;
}

/**
 * 处理请求
 */
static int easy_connection_do_request(easy_message_t *m)
{
    easy_connection_t       *c;
    void                    *packet;
    easy_request_t          *r, *rn;
    int                     cnt, ret;

    cnt = 0;
    c = m->c;

    // 处理buf, decode
    while (m->input->pos < m->input->last) {
        if ((packet = (c->handler->decode)(m)) == NULL) {
            if (m->status != EASY_ERROR)
                break;

            easy_warn_log("decode error, m=%p, fd=%d\n", m, c->fd);
            return EASY_ERROR;
        }

        // new request
        r = (easy_request_t *)easy_pool_calloc(m->pool, sizeof(easy_request_t));
        r->ms = (easy_message_session_t *)m;
        r->ipacket = packet;    //进来的数据包

        // add m->request_list
        easy_list_add_tail(&r->request_list_node, &m->request_list);
        cnt ++;
    }

    // cnt
    if (cnt) {
        m->request_list_count += cnt;
        c->doing_request_count += cnt;
        easy_atomic32_add(&c->ioth->doing_request_count, cnt);
        m->recycle_cnt ++;
    }

    if ((m = easy_connection_recycle_message(m)) == NULL)
        return EASY_ERROR;

    m->status = ((m->input->pos < m->input->last) ? EASY_MESG_READ_AGAIN : 0);


    // batch process
    if (c->handler->batch_process)
        (c->handler->batch_process)(m);

    // process
    cnt = 0;
    easy_list_for_each_entry_safe(r, rn, &m->request_list, request_list_node) {
        easy_list_del(&r->request_list_node);
        EASY_IOTH_SELF->done_request_count ++;

        // process
        if ((ret = (c->handler->process)(r)) == EASY_ERROR)
            return EASY_ERROR;

        if (ret == EASY_OK && easy_connection_request_done(r) == EASY_OK) {
            cnt ++;
        }

        // write to socket
        if (cnt >= 128) {
            cnt = 0;

            if (easy_connection_write_socket(c) == EASY_ABORT)
                return EASY_ERROR;
        }
    }

    // 所有的request都有reply了,一起才响应
    if (easy_connection_write_socket(c) == EASY_ABORT) {
        return EASY_ERROR;
    }

    if (m->request_list_count == 0 && m->status != EASY_MESG_READ_AGAIN) {
        easy_message_destroy(m, 1);
    }

    // 加入监听
    if (c->event_status == EASY_EVENT_READ && !c->wait_close)
        easy_connection_evio_start(c);

    easy_connection_redispatch_thread(c);

    return EASY_OK;
}

/**
 * write事件处理
 */
static void easy_connection_on_writable(struct ev_loop *loop, ev_io *w, int revents)
{
    easy_connection_t       *c;
    int                     ret;

    c = (easy_connection_t *)w->data;
    assert(c->fd == w->fd);

    ret = easy_connection_write_socket(c);

    if (ret == EASY_ABORT)
        goto error_exit;

    // connected.
    if (c->status == EASY_CONN_CONNECTING && c->type == EASY_TYPE_CLIENT) {
        c->status = EASY_CONN_OK;
        ev_io_start(c->loop, &c->read_watcher);
        ev_timer_set(&c->timeout_watcher, 0.0, EASY_MIN_INTERVAL);
        ev_timer_again(c->loop, &c->timeout_watcher);

        // on connect
        if (c->handler->on_connect)
            (c->handler->on_connect)(c);
    }

    // send new packet
    if (c->handler->new_packet && c->type == EASY_TYPE_CLIENT && ret == EASY_OK
            && c->doing_request_count < (EASY_CONN_DOING_REQ_CNT >> 1)) {
        if ((c->handler->new_packet)(c) == EASY_ERROR)
            goto error_exit;
    }

    // 没数据可发, 把write停掉
    if (easy_list_empty(&c->output))
        ev_io_stop(c->loop, &c->write_watcher);

    return;
error_exit:
    easy_connection_destroy(c);
}

/**
 * 对timeout的处理message
 */
static void easy_connection_on_timeout_mesg(struct ev_loop *loop, ev_timer *w, int revents)
{
    static int now = 0;
    easy_connection_t   *c;
    easy_session_t      *s;

    s = (easy_session_t *)w->data;
    c = s->c;

    easy_hash_dlist_del(c->send_queue, s->packet_id);
    s->packet_id = 0;

    // process
    if (now != (int)ev_now(loop)) {
        easy_warn_log("timeout_mesg: %p\n", s);
        now = (int)ev_now(loop);
    }

    if (easy_session_process(s, 1) == EASY_ERROR)
        easy_connection_destroy(c);
}

/**
 * 对timeout的处理connection
 */
static void easy_connection_on_timeout_conn(struct ev_loop *loop, ev_timer *w, int revents)
{
    easy_session_t          *s, *s2;
    easy_connection_t       *c;
    double                  t, now;
    c = (easy_connection_t *)w->data;

    easy_debug_log("timeout_conn: c=%p, status=%d, type=%d\n", c, c->status, c->type);

    if (c->status == EASY_CONN_AUTO_CONN) {
        easy_connection_autoconn(c);
    } else if (c->status != EASY_CONN_OK) { // EASY_CONN_CLOSE, EASY_CONN_CONNECTING
        easy_connection_destroy(c);
    } else { // EASY_CONN_OK
        now = ev_now(loop);

        if (c->handler->on_idle && c->last_time + c->idle_time / 1000.0 < now) {
            c->last_time = now;
            (c->handler->on_idle)(c);
        }

        if (c->type == EASY_TYPE_CLIENT) {
            if (easy_list_empty(&c->session_list)) {
                if (c->handler->on_idle) {
                    t = easy_max(1.0, c->idle_time / 2000.0);
                    ev_timer_set(&c->timeout_watcher, 0.0, t);
                    ev_timer_again(c->loop, &c->timeout_watcher);
                } else {
                    ev_timer_stop(c->loop, &c->timeout_watcher);
                }
            }

            // 延迟开始超时
            easy_list_for_each_entry_safe(s, s2, &c->session_list, session_list_node) {
                t = s->timeout_watcher.at - now;

                if (t > EASY_MIN_INTERVAL) break;

                easy_list_del(&s->session_list_node);
                // 超时
                s->timeout_watcher.at = t;
                ev_timer_start(loop, &s->timeout_watcher);
            }
        }
    }
}

/**
 * 把connection上的output的buffer写到socket上
 *
 * @param c - easy_connection_t对象
 * @return  - EASY_ABORT 网络断开
 *            EASY_AGAIN 没写完,需要继续写
 *            EASY_OK    写完了
 */
int easy_connection_write_socket(easy_connection_t *c)
{
    int ret;

    // 空的直接返回
    if (easy_list_empty(&c->output))
        return EASY_OK;

    // 加塞
    if (EASY_IOTH_SELF->eio->tcp_cork && c->tcp_cork_flag == 0) {
        easy_socket_set_tcpopt(c->fd, TCP_CORK, 1);
        c->tcp_cork_flag = 1;
    }

    ret = easy_socket_write(c->fd, &c->output);

    if (ret == EASY_ERROR) {
        char buffer[32];
        easy_warn_log("ret=%d, addr: %s, error: %s (%d)\n", ret,
                      easy_inet_addr_to_str(&c->addr, buffer, 32), strerror(errno), errno);
        c->conn_has_error = 1;
        return EASY_ABORT;
    }

    c->last_time = ev_now(c->loop);

    return easy_connection_write_again(c);
}

// 判断write again
static int easy_connection_write_again(easy_connection_t *c)
{
    easy_request_t      *r, *rn;

    // 还有没写出去, 起写事件
    if (easy_list_empty(&c->output) == 0) {
        ev_io_start(c->loop, &c->write_watcher);
        return EASY_AGAIN;
    } else {
        if (c->type == EASY_TYPE_SERVER) {
            if (!easy_list_empty(&c->session_list)) {
                easy_list_for_each_entry_safe(r, rn, &c->session_list, request_list_node) {
                    easy_list_del(&r->request_list_node);
                    (c->handler->process)(r);
                }
            } else if (c->wait_close) {// 需要关闭掉
                return EASY_ABORT;
            }
        }

        // tcp_nodelay, tcp_cork
        if (EASY_IOTH_SELF->eio->tcp_nodelay && c->tcp_nodelay_flag == 0) {
            easy_socket_set_tcpopt(c->fd, TCP_NODELAY, 1);
            c->tcp_nodelay_flag = 1;
        } else if (EASY_IOTH_SELF->eio->tcp_cork && c->tcp_cork_flag) {
            easy_socket_set_tcpopt(c->fd, TCP_CORK, 0);
            c->tcp_cork_flag = 0;
        }
    }

    return EASY_OK;
}

/**
 * 得到packet的id
 */
static uint64_t easy_connection_get_packet_id(easy_connection_t *c, void *packet, int flag)
{
    uint64_t packet_id = 0;

    if (c->handler->get_packet_id) {
        packet_id = (c->handler->get_packet_id)(c, packet);
    } else {
        packet_id = c->send_queue->seqno;

        if (flag) packet_id -= c->send_queue->count;

        packet_id <<= 16;
        packet_id |= (c->fd & 0xffff);
    }

    return packet_id;
}

/**
 * new 出一个connection_t对象
 */
static easy_connection_t *easy_connection_new()
{
    easy_pool_t             *pool;
    easy_connection_t       *c;

    // 为connection建pool
    if ((pool = easy_pool_create(0)) == NULL)
        return NULL;

    // 创建easy_connection_t对象
    c = (easy_connection_t *) easy_pool_calloc(pool, sizeof(easy_connection_t));

    if (c == NULL)
        goto error_exit;

    // 初始化
    c->pool = pool;
    c->sw_time = 0.0;
    c->reconn_time = 100;
    c->idle_time = 60000;
    easy_list_init(&c->message_list);
    easy_list_init(&c->session_list);
    easy_list_init(&c->conn_list_node);
    easy_list_init(&c->output);

    return c;
error_exit:
    easy_pool_destroy(pool);
    return NULL;
}

/**
 * server回复
 */
static int easy_connection_send_response(easy_list_t *request_list)
{
    easy_request_t              *r, *rn;
    easy_message_t              *m;
    easy_connection_t           *c, *nc;
    easy_connection_list_t      wlist = {NULL, NULL};
    easy_connection_list_t      flist = {NULL, NULL};
    int                         ret;

    // encode
    easy_list_for_each_entry_safe(r, rn, request_list, request_list_node) {
        easy_list_del(&r->request_list_node);
        m = (easy_message_t *) r->ms;
        c = m->c;

        // 从其他进程返回后
        ret = EASY_ERROR;

        if (r->retcode != EASY_ERROR) {
            if ((ret = easy_connection_request_done(r)) == EASY_OK) {
                easy_connection_add_list(&wlist, c);
            }
        } else { // 如果出错
            easy_connection_add_list(&flist, c);
        }

        // 引用计数
        c->pool->ref --;
        easy_atomic_dec(&m->pool->ref);

        // message是否也不在使用了
        if (m->request_list_count == 0 && ret == EASY_OK && m->status != EASY_MESG_READ_AGAIN)
            easy_message_destroy(m, 1);
    }

    // failure request, close connection
    if (flist.tail) {
        flist.tail->next = NULL;
        nc = flist.head;

        while((c = nc)) {
            nc = c->next;
            c->next = NULL;
            easy_connection_destroy(c);
        }
    }

    // foreach write socket
    if (wlist.tail) {
        wlist.tail->next = NULL;
        nc = wlist.head;

        while((c = nc)) {
            nc = c->next;
            c->next = NULL;

            if (easy_connection_write_socket(c) == EASY_ABORT) {
                easy_connection_destroy(c);
            } else if (c->type == EASY_TYPE_SERVER) {
                easy_connection_redispatch_thread(c);
            }
        }
    }

    return EASY_OK;
}

int easy_connection_send_session_list(easy_list_t *list)
{
    easy_connection_t       *c;
    easy_session_t          *s, *s1;
    easy_connection_list_t  wlist = {NULL, NULL};
    int                     status;

    // foreach encode
    easy_list_for_each_entry_safe(s, s1, list, session_list_node) {
        easy_list_del(&s->session_list_node);

        // connect, disconnect
        status = s->status;

        if ((c = easy_connection_do_client(s)) == NULL || status)
            continue;

        // build session
        s->c = c;

        if (easy_connection_session_build(s) == EASY_OK) {
            easy_connection_add_list(&wlist, s->c);
        }
    }

    // foreach
    if (wlist.tail)
        wlist.tail->next = NULL;

    while(wlist.head) {
        c = wlist.head;
        wlist.head = c->next;
        c->next = NULL;

        if (c->status != EASY_CONN_OK || ev_is_active(&c->write_watcher))
            continue;

        if (easy_connection_write_socket(c) == EASY_ABORT) {
            easy_connection_destroy(c);
        }
    }

    return EASY_OK;
}

static void easy_connection_add_list(easy_connection_list_t *wlist, easy_connection_t *c)
{
    if (c->next == NULL) {
        if (wlist->head == NULL)
            wlist->head = c;
        else
            wlist->tail->next = c;

        wlist->tail = c;
        c->next = c;
    }
}

static easy_message_t *easy_connection_recycle_message(easy_message_t *m)
{
    easy_message_t          *newm;
    int                     len;

    len = (m->input->last - m->input->pos);

    if (m->recycle_cnt < 16 || len == 0)
        return m;

    // 增加default_message_len大小
    m->c->default_message_len += len;
    newm = easy_message_create(m->c);
    m->c->default_message_len -= len;

    if (newm == NULL)
        return NULL;

    // 把旧的移到新的上面
    memcpy(newm->input->pos, m->input->pos, len);
    newm->input->last += len;

    // 删除之前的message
    if (m->request_list_count == 0) {
        easy_message_destroy(m, 1);
    } else {
        m->status = 0;
    }

    return newm;
}

/**
 * 连接到addrv
 */
static easy_connection_t *easy_connection_do_connect(easy_client_t *client)
{
    struct sockaddr         addr;
    int                     fd, v;
    easy_connection_t       *c;
    double                  t;
    char                    buffer[32];

    // 建立一个connection
    if ((c = easy_connection_new()) == NULL) {
        easy_error_log("new connect failure.\n");
        return NULL;
    }

    if ((fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        easy_error_log("socket failure: %s (%d)\n", strerror(errno), errno);
        goto error_exit;
    }

    // 初始化
    c->fd = fd;
    c->type = EASY_TYPE_CLIENT;
    c->handler = client->handler;
    c->addr = client->addr;
    c->client = client;

    // 连接
    memset(&addr, 0, sizeof(addr));
    memcpy(&addr, &client->addr, sizeof(uint64_t));
    easy_socket_non_blocking(fd);

    if (connect(fd, &addr, sizeof(struct sockaddr)) < 0) {
        if (errno != EINPROGRESS) {
            easy_error_log("connect to '%s' failure: %s (%d)\n",
                           easy_inet_addr_to_str(&c->addr, buffer, 32), strerror(errno), errno);
            goto error_exit;
        }

        c->status = EASY_CONN_CONNECTING;
    }

    v = offsetof(easy_session_t, send_queue_hash);
    c->send_queue = easy_hash_create(c->pool, EASY_CONN_DOING_REQ_CNT, v);

    // 初始化事件
    ev_io_init(&c->read_watcher, easy_connection_on_readable, fd, EV_READ);
    ev_io_init(&c->write_watcher, easy_connection_on_writable, fd, EV_WRITE);
    t = (client->timeout ? client->timeout : EASY_CLIENT_DEFAULT_TIMEOUT) / 1000.0;
    ev_timer_init(&c->timeout_watcher, easy_connection_on_timeout_conn, t, 0.0);
    c->read_watcher.data = c;
    c->write_watcher.data = c;
    c->timeout_watcher.data = c;

    // event_status
    if (c->status == EASY_CONN_CONNECTING)
        v = (EASY_EVENT_TIMEOUT | EASY_EVENT_WRITE);
    else
        v = (EASY_EVENT_TIMEOUT | EASY_EVENT_READ);

    easy_debug_log("connect to '%s' success, connection=%p, fd=%d\n",
                   easy_inet_addr_to_str(&c->addr, buffer, 32), c, c->fd);

    c->event_status = v;
    c->ioth = EASY_IOTH_SELF;
    c->loop = c->ioth->loop;

    // 加入
    easy_list_add_tail(&c->conn_list_node, &c->ioth->connected_list);
    easy_connection_evio_start(c);

    return c;
error_exit:

    if (fd >= 0) close(fd);

    easy_pool_destroy(c->pool);
    return NULL;
}

static easy_connection_t *easy_connection_do_client(easy_session_t *s)
{
    easy_io_thread_t        *ioth = EASY_IOTH_SELF;
    easy_connection_t       *c = NULL;
    easy_client_t           *client;

    // hashcode
    client = (easy_client_t *)easy_client_list_find(ioth->client_list, &s->addr);

    // 正常的session
    if (s->status == 0) {
        c = (client && client->ref ? client->c : NULL);

        if (s->process == NULL && c && c->handler) {
            s->process = c->handler->process;
        }

        if (c == NULL || ioth->eio->stoped) {
            s->error = 1;
            c = NULL;
            easy_session_process(s, 0);
        }

        return c;
        // 连接指令
    } else if (s->status == EASY_CONNECT_ADDR) {
        if (client == NULL) {
            if ((client = (easy_client_t *)easy_array_alloc(ioth->client_array)) == NULL)
                goto error_exit;

            memset(client, 0, sizeof(easy_client_t));
            client->addr = s->addr;
            client->handler = (easy_io_handler_pt *)s->process;
            client->timeout = s->timeout;
            client->user_data = s->r.args;
            easy_client_list_add(ioth->client_list, &client->addr, &client->client_list_node);
        }

        if (client->c == NULL && (client->c = easy_connection_do_connect(client)) == NULL)
            goto error_exit;

        c = client->c;
        c->user_data = client->user_data;

        if (s->packet_id) c->auto_reconn = 1;

        client->ref ++;
        // 断开指令
    } else if (client && --client->ref == 0) {
        if (client->c) {
            client->c->wait_close = 1;
            client->c->client = NULL;
        }

        easy_hash_del_node(&client->client_list_node);
        easy_array_free(ioth->client_array, client);
    }

error_exit:

    if (s->pool) easy_pool_destroy(s->pool);

    return c;
}

int easy_connection_request_done(easy_request_t *r)
{
    easy_connection_t       *c;
    easy_message_t          *m;
    int                     retcode = r->retcode;

    m = (easy_message_t *) r->ms;
    c = m->c;

    // encode
    if (r->opacket) {
        if ((c->handler->encode)(r, r->opacket) != EASY_OK)
            return EASY_ERROR;

        if (retcode == EASY_AGAIN) { // 当write_socket写完
            easy_list_add_tail(&r->request_list_node, &c->session_list);
            r->opacket = NULL;
        }
    }

    // retcode
    if (retcode == EASY_OK) {
        // 计算
        assert(m->request_list_count > 0);
        assert(c->doing_request_count > 0);
        m->request_list_count --;
        c->doing_request_count --;
        c->done_request_count ++;
        easy_atomic32_dec(&c->ioth->doing_request_count);
        easy_list_add_tail(&r->request_list_node, &m->request_done_list);

        // 设置redispatch
        if (unlikely((c->done_request_count & 0xfff) == 8 && c->sw_time < (ev_now(c->loop) - 30))) {
            c->need_redispatch = 1;
            c->sw_time = ev_now(c->loop);
        }
    }

    return EASY_OK;
}

static void easy_connection_autoconn(easy_connection_t *c)
{
    int                 fd;
    struct sockaddr     addr;
    char                buffer[32];

    c->status = EASY_CONN_CLOSE;

    if (c->client == NULL)
        return;

    if ((fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        easy_error_log("socket failure: %s (%d)\n", strerror(errno), errno);
        goto error_exit;
    }

    easy_socket_non_blocking(fd);
    memset(&addr, 0, sizeof(addr));
    memcpy(&addr, &c->addr, sizeof(uint64_t));

    if (connect(fd, &addr, sizeof(struct sockaddr)) < 0) {
        if (errno == ECONNREFUSED) {
            easy_error_log("connect to '%s' failure: %s (%d)\n",
                           easy_inet_addr_to_str(&c->addr, buffer, 32), strerror(errno), errno);
            close(fd);
            return;
        }

        if (errno != EINPROGRESS) {
            easy_error_log("connect to '%s' failure: %s (%d)\n",
                           easy_inet_addr_to_str(&c->addr, buffer, 32), strerror(errno), errno);
            goto error_exit;
        }

        c->status = EASY_CONN_CONNECTING;
    } else {
        c->status = EASY_CONN_OK;
    }

    // 初始化
    c->fd = fd;
    ev_io_set(&c->read_watcher, fd, EV_READ);
    ev_io_set(&c->write_watcher, fd, EV_WRITE);

    if (c->status == EASY_CONN_CONNECTING)
        c->event_status = (EASY_EVENT_TIMEOUT | EASY_EVENT_WRITE);
    else
        c->event_status = (EASY_EVENT_TIMEOUT | EASY_EVENT_READ);

    easy_debug_log("reconnect to '%s' success, connection=%p, fd=%d\n",
                   easy_inet_addr_to_str(&c->addr, buffer, 32), c, c->fd);

    easy_connection_evio_start(c);
    return;
error_exit:
    c->auto_reconn = 0;
    easy_connection_destroy(c);
}
